package com.atguigu.flink.datastreamapi.agg;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/4/4
 *
 *  Reduce的特点：
 *      ①两两聚合
 *      ②输入和输出类型必须一样的
 */
public class Demo2_Reduce
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        KeyedStream<WaterSensor, String> ds = env.socketTextStream("hadoop102", 8888)
                                                        .map(new WaterSensorMapFunction())
                                                        .keyBy(WaterSensor::getId);


        ds.reduce(new ReduceFunction<WaterSensor>()
        {
            /*
                每两个运算一次
                WaterSensor value1： 上一次聚合的结果
                WaterSensor value2： 新到达的数据
             */
            @Override
            public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                System.out.println("Demo2_Reduce.reduce");
                //求vc最大值，其他列要求输出最大值的数据的字段
                if (value2.getVc() >= value1.getVc()){
                    return value2;
                }else {
                    return value1;
                }
            }
        })
          .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
